package com.heiye.comment.biz.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson.JSON;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.heiye.comment.biz.constant.MQConstants;
import com.heiye.comment.biz.constant.RedisKeyConstants;
import com.heiye.comment.biz.domain.dataobject.CommentDO;
import com.heiye.comment.biz.domain.dataobject.CommentLikeDO;
import com.heiye.comment.biz.domain.mapper.CommentDOMapper;
import com.heiye.comment.biz.domain.mapper.CommentLikeDOMapper;
import com.heiye.comment.biz.enums.CommentLikeLuaResultEnum;
import com.heiye.comment.biz.enums.CommentUnlikeLuaResultEnum;
import com.heiye.comment.biz.enums.LikeUnlikeCommentTypeEnum;
import com.heiye.comment.biz.enums.ResponseCodeEnum;
import com.heiye.comment.biz.model.dto.LikeUnlikeCommentMqDTO;
import com.heiye.comment.biz.model.dto.PublishCommentMqDTO;
import com.heiye.comment.biz.model.vo.*;
import com.heiye.comment.biz.retry.SendMqRetryHelper;
import com.heiye.comment.biz.rpc.*;
import com.heiye.comment.biz.service.CommentService;
import com.heiye.count.api.dto.FindCommentCountByIdRspDTO;
import com.heiye.framework.common.constant.DateConstants;
import com.heiye.framework.common.exception.BizException;
import com.heiye.framework.common.response.PageResponse;
import com.heiye.framework.common.response.Response;
import com.heiye.framework.common.util.DateUtils;
import com.heiye.framework.common.util.JsonUtils;
import com.heiye.kv.api.dto.req.FindCommentContentReqDTO;
import com.heiye.kv.api.dto.rsp.FindCommentContentRspDTO;
import com.heiye.user.api.dto.resp.FindUserByIdRspDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.util.Strings;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.heiye.framework.biz.context.thread.LoginUserContextHolder;
import org.springframework.core.io.ClassPathResource;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;
import org.springframework.transaction.support.TransactionTemplate;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author: heiye
 * @date: 2025/07/24 下午3:28
 * @version: v1.0.0
 * @description: 评论业务
 */
@Slf4j
@Service
public class CommentServiceImpl implements CommentService {

    @Resource
    private SendMqRetryHelper sendMqRetryHelper;

    @Resource
    private DistributedIdGeneratorRpcService distributedIdGeneratorRpcService;

    @Resource
    private CommentDOMapper commentDOMapper;

    @Resource
    private KeyValueRpcService keyValueRpcService;

    @Resource
    private UserRpcService userRpcService;

    @Resource(name = "taskExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Resource
    private CommentLikeDOMapper commentLikeDOMapper;

    @Resource
    private TransactionTemplate transactionTemplate;

    @Resource
    private CountRpcService countRpcService;

    @Resource
    private QuizRpcService quizRpcService;

    /**
     * 评论详情本地缓存
     */
    private static final Cache<Long, String> LOCAL_CACHE = Caffeine.newBuilder()
            // 设置初始容量为 10000 个条目
            .initialCapacity(10000)
            // 设置缓存的最大容量为 10000 个条目
            .maximumSize(10000)
            // 设置缓存条目在写入后 1 小时过期
            .expireAfterWrite(1, TimeUnit.HOURS)
            .build();

    /**
     * 发布评论
     *
     * @param publishCommentReqVO
     * @return
     */
    @Override
    public Response<?> publishComment(PublishCommentReqVO publishCommentReqVO) {
        // 评论正文
        String content = publishCommentReqVO.getContent();
        // 评论图片
        String imageUrl = publishCommentReqVO.getImageUrl();

        // 评论内容和图片不能同时为空
        Preconditions.checkArgument(StringUtils.isNotBlank(content) || StringUtils.isNotBlank(imageUrl),
                "评论正文和图片不能同时为空");

        // 发布者 ID
        Long creatorId = LoginUserContextHolder.getUserId();

        // RPC: 调用分布式 ID 生成服务,  生成评论 ID
        String commentId = distributedIdGeneratorRpcService.generateCommentId();

        // 发送 MQ
        // 构建消息体 DTO
        PublishCommentMqDTO publishCommentMqDTO = PublishCommentMqDTO.builder()
                .quizId(publishCommentReqVO.getQuizId())
                .commentId(Long.valueOf(commentId))
                .content(content)
                .imageUrl(imageUrl)
                .creatorId(creatorId)
                .replyCommentId(publishCommentReqVO.getReplyCommentId())
                .createTime(LocalDateTime.now())
                .build();

        // 发送 MQ (包含重试机制)
        sendMqRetryHelper.asyncSend(MQConstants.TOPIC_PUBLISH_COMMENT, JsonUtils.toJsonString(publishCommentMqDTO));

        return Response.success();
    }

    /**
     * 评论列表游标分页查询
     *
     * @param findCommentCursorPageListReqVO
     * @return
     */
    @Override
    public Response<FindCommentCursorPageListRspVO> findCommentPageList(FindCommentCursorPageListReqVO findCommentCursorPageListReqVO) {
        // 问题 ID
        Long quizId = findCommentCursorPageListReqVO.getQuizId();
        // 游标
        Long cursor = findCommentCursorPageListReqVO.getCursor();
        // 当前页码
        Integer pageNo = findCommentCursorPageListReqVO.getPageNo();

        // 计算分页查询的偏移量 offset
        long offset = PageResponse.getOffset(pageNo, 10);

        // 评论分页缓存使用 ZSET + STRING 实现
        // 构建评论 ZSET Key
        String commentZSetKey = RedisKeyConstants.buildCommentListKey(quizId);
        // 先判断 ZSET 是否存在
        boolean hasKey = redisTemplate.hasKey(commentZSetKey);

        // 若不存在
        if (!hasKey) {
            // 异步将评论同步到 redis 中 (最多同步 500 条)
            threadPoolTaskExecutor.execute(() ->
                    syncComments2Redis(commentZSetKey, quizId));
        }

        // 如果是第 1 页，需要查询该问题下是否存在最佳回答
        Long bestCommentId = null;
        FindCommentItemRspVO bestComment = null;
        if (pageNo == 1) {
            bestCommentId = getBestCommentId(quizId);
            // 最佳评论 ID 不为空, 则查询最佳评论
            if (Objects.nonNull(bestCommentId)) {
                bestComment = getBestCommentById(bestCommentId, quizId);
            }
        }

        // 若 ZSET 缓存存在, 并且查询的是前 5 页的评论
        if (hasKey && offset < 50) {
            // 使用 ZRevRange 获取某问题下，按时间降序排序的评论 ID
            Set<Object> commentIds = redisTemplate.opsForZSet()
                    .reverseRangeByScore(commentZSetKey, -Double.MAX_VALUE, Double.MAX_VALUE, offset, 10);

            // 最终返回返回参数
            List<FindCommentItemRspVO> findCommentItemRspVOS = Lists.newArrayList();

            // 若结果不为空
            if (CollUtil.isNotEmpty(commentIds)) {
                // Set 转 List
                ArrayList<Object> commentIdList = Lists.newArrayList(commentIds);

                // 先查询本地缓存
                // 新建一个集合，用于存储本地缓存中不存在的评论 ID
                List<Long> localCacheExpiredCommentIds = Lists.newArrayList();

                // 构建查询本地缓存的 Key 集合
                List<Long> localCacheKeys = commentIds.stream()
                        .map(commentId -> Long.valueOf(commentId.toString()))
                        .toList();

                // 批量查询本地缓存
                Map<Long, String> commentIdAndDetailJsonMap = LOCAL_CACHE.getAll(localCacheKeys, missingKeys -> {
                    // 对于本地缓存中缺失的 key，返回空字符串
                    Map<Long, String> missingData = Maps.newHashMap();
                    missingKeys.forEach(key -> {
                        // 记录缓存中不存在的评论 ID
                        localCacheExpiredCommentIds.add(key);
                        // 不存在的评论详情, 对其 Value 值设置为空字符串
                        missingData.put(key, Strings.EMPTY);
                    });
                    return missingData;
                });

                // 若 localCacheExpiredCommentIds 的大小不等于 commentIdList 的大小，说明本地缓存中有数据
                if (localCacheExpiredCommentIds.size() != commentIdList.size()) {
                    // 将本地缓存中的评论详情 Json, 转换为实体类，添加到 VO 返参集合中
                    for (String value : commentIdAndDetailJsonMap.values()) {
                        if (StringUtils.isBlank(value)) {
                            continue;
                        }
                        FindCommentItemRspVO findCommentItemRspVO = JsonUtils.parseObject(value, FindCommentItemRspVO.class);
                        findCommentItemRspVOS.add(findCommentItemRspVO);
                    }
                }

                // 若 localCacheExpiredCommentIds 大小等于 0，说明评论详情数据都在本地缓存中，直接响应返参
                if (CollUtil.size(localCacheExpiredCommentIds) == 0) {

                    // 设置评论计数数据和用户是否点赞信息
                    setCommentCountAndLikeUnLikeInfo(findCommentItemRspVOS);

                    // 添加最佳评论，并设置游标，最后反参
                    FindCommentCursorPageListRspVO findCommentCursorPageListRspVO = addBestCommentAndSetCursor(bestComment, findCommentItemRspVOS);

                    return PageResponse.success(findCommentCursorPageListRspVO);
                }

                // 构建 MGET 批量查询评论详情的 Key 集合
                List<String> commentIdKeys = localCacheExpiredCommentIds.stream()
                        .map(RedisKeyConstants::buildCommentDetailKey)
                        .toList();
                // MGET 批量查询评论详情的 Key 集合
                List<Object> commentsJsonList = redisTemplate.opsForValue().multiGet(commentIdKeys);

                // 可能存在部分评论不在缓存中，已经过期被删除，这些评论 ID 需要提取出来，等会查数据库
                List<Long> expiredCommentIds = Lists.newArrayList();
                for (int i = 0; i < commentsJsonList.size(); i++) {
                    String commentJson = (String) commentsJsonList.get(i);
                    Long commentId = Long.valueOf(localCacheExpiredCommentIds.get(i).toString());
                    if (Objects.nonNull(commentJson)) {
                        // 缓存中存在的评论 Json，直接转换为 VO 添加到返参集合中
                        FindCommentItemRspVO findCommentItemRspVO = JsonUtils.parseObject(commentJson, FindCommentItemRspVO.class);
                        findCommentItemRspVOS.add(findCommentItemRspVO);
                    } else {
                        // 评论失效，添加到失效评论列表
                        expiredCommentIds.add(commentId);
                    }
                }

                // 对于不存在的评论，需要批量从数据库中查询，并添加到 commentRspVOS 中
                if (CollUtil.isNotEmpty(expiredCommentIds)) {
                    List<CommentDO> commentDOS = commentDOMapper.selectByCommentId(expiredCommentIds);
                    // 获取数据库查询出的评论详情
                    List<FindCommentItemRspVO> findCommentItemRspVOS1 = getCommentDataAndSync2Redis(commentDOS, quizId);

                    // 添加到返参集合中
                    findCommentItemRspVOS.addAll(findCommentItemRspVOS1);

                    // 按时间进行降序排列
                    findCommentItemRspVOS = findCommentItemRspVOS.stream()
                            .sorted(Comparator.comparing(FindCommentItemRspVO::getCommentId).reversed())
                            .collect(Collectors.toList());
                }

                // 异步将评论详情，同步到本地缓存
                syncCommentDetail2LocalCache(findCommentItemRspVOS);

                // 设置评论计数数据和用户是否点赞信息
                setCommentCountAndLikeUnLikeInfo(findCommentItemRspVOS);
            }

            // 添加最佳评论，并设置游标，最后反参
            FindCommentCursorPageListRspVO findCommentCursorPageListRspVO = addBestCommentAndSetCursor(bestComment, findCommentItemRspVOS);

            return PageResponse.success(findCommentCursorPageListRspVO);
        }

        // 缓存中没有，则查询数据库
        List<CommentDO> commentDOS = commentDOMapper.selectCursorPageList(quizId, cursor, bestCommentId);

        // 构建返回参数
        List<FindCommentItemRspVO> findCommentItemRspVOS = Lists.newArrayList();

        if (CollUtil.isNotEmpty(commentDOS)) {
            findCommentItemRspVOS = getCommentDataAndSync2Redis(commentDOS, quizId);
            // 异步将评论详情，同步到本地缓存
            syncCommentDetail2LocalCache(findCommentItemRspVOS);

            // 设置评论计数数据和用户是否点赞信息
            setCommentCountAndLikeUnLikeInfo(findCommentItemRspVOS);
        }

        // 添加最佳评论，并设置游标，最后反参
        FindCommentCursorPageListRspVO findCommentCursorPageListRspVO = addBestCommentAndSetCursor(bestComment, findCommentItemRspVOS);

        return Response.success(findCommentCursorPageListRspVO);
    }

    /**
     * 根据最佳评论 ID 查询最佳评论
     *
     * @param bestCommentId
     * @return
     */
    private FindCommentItemRspVO getBestCommentById(Long bestCommentId, Long quizId) {
        // 先从本地缓存中查询
        String bestCommentIdLocalCache = LOCAL_CACHE.getIfPresent(bestCommentId);

        // 如果本地缓存在数据，转换类型后返回数据
        if (Objects.nonNull(bestCommentIdLocalCache)) {
            // 转换类型
            FindCommentItemRspVO findCommentItemRspVO = JsonUtils.parseObject(bestCommentIdLocalCache, FindCommentItemRspVO.class);

            // 设置评论计数数据和用户是否点赞信息
            setCommentCountAndLikeUnLikeInfo(Lists.newArrayList(findCommentItemRspVO));

            return findCommentItemRspVO;
        }

        // 否则，再从 Redis 中查询
        String commentDetailRedisKey = RedisKeyConstants.buildCommentDetailKey(bestCommentId);

        // 判断 Redis 中是否有缓存
        boolean commentDetailRedisHasKey = redisTemplate.hasKey(commentDetailRedisKey);

        // 如果Redis 中有缓存，取出后转换类型返回数据
        if (commentDetailRedisHasKey) {
            String commentDetailRedisValue = (String) redisTemplate.opsForValue().get(commentDetailRedisKey);

            if (Objects.nonNull(commentDetailRedisValue)) {
                // 将数据同步到本地缓存中
                LOCAL_CACHE.put(bestCommentId, commentDetailRedisValue);
                // 转换类型
                FindCommentItemRspVO findCommentItemRspVO = JsonUtils.parseObject(commentDetailRedisValue, FindCommentItemRspVO.class);

                // 设置评论计数数据和用户是否点赞信息
                setCommentCountAndLikeUnLikeInfo(Lists.newArrayList(findCommentItemRspVO));

                return findCommentItemRspVO;
            }
        }

        // 如果缓存中都没有数据，则从数据库中查询
        CommentDO commentDO = commentDOMapper.selectByPrimaryKey(bestCommentId);

        // 获取数据库查询出的评论详情
        List<FindCommentItemRspVO> findCommentItemRspVOList = getCommentDataAndSync2Redis(Lists.newArrayList(commentDO), quizId);

        // 异步将评论详情，同步到本地缓存
        syncCommentDetail2LocalCache(findCommentItemRspVOList);

        // 设置评论计数数据和用户是否点赞信息
        setCommentCountAndLikeUnLikeInfo(findCommentItemRspVOList);

        return findCommentItemRspVOList.get(0);
    }

    /**
     * 添加最佳评论，并设置游标，最后反参
     *
     * @param bestComment
     * @param findCommentItemRspVOS
     * @return
     */
    private static FindCommentCursorPageListRspVO addBestCommentAndSetCursor(FindCommentItemRspVO bestComment, List<FindCommentItemRspVO> findCommentItemRspVOS) {
        // 如果存在最佳评论，则将评论添加至列表中并进行一次排除
        if (Objects.nonNull(bestComment)) {
            Long bestCommentId = bestComment.getCommentId();
            // 排除数据库中可能查询出最佳评论
            findCommentItemRspVOS = findCommentItemRspVOS.stream()
                    .filter(findCommentItemRspVO -> !findCommentItemRspVO.getCommentId().equals(bestCommentId))
                    .collect(Collectors.toList());
            // 添加最佳评论
            findCommentItemRspVOS.add(0, bestComment);
        }

        // 过滤出最早发布的问题 ID，充当下一页的游标
        Optional<Long> earliestCommentId = findCommentItemRspVOS.stream().map(FindCommentItemRspVO::getCommentId).min(Long::compareTo);

        return FindCommentCursorPageListRspVO.builder()
                .commentList(findCommentItemRspVOS)
                .cursor(earliestCommentId.orElse(null))
                .build();
    }

    /**
     * 获取最佳评论ID
     *
     * @param quizId
     * @return
     */
    private Long getBestCommentId(Long quizId) {
        String commentBestRedisKey = RedisKeyConstants.buildCommentBestKey(quizId);
        // 查询 Redis 是否有缓存最佳评论的 ID
        boolean commentBestRedisHasKey = redisTemplate.hasKey(commentBestRedisKey);
        // 如果存在，则获取该 ID，并查询该评论在缓存中存放的评论详情
        if (commentBestRedisHasKey) {
            Object redisValue = redisTemplate.opsForValue().get(commentBestRedisKey);

            if (Objects.nonNull(redisValue)) {
                return Long.valueOf(redisValue.toString());
            }
        }

        // 不存在，则查询该问题下是否存在最佳评论
        CommentDO bestComment = commentDOMapper.selectBestCommentByQuizId(quizId);

        // 如果存在最佳评论，将评论ID 存入缓存中
        if (Objects.nonNull(bestComment)) {
            // 存入缓存中并查询对应数据
            // 随机过期时间 5 小时 +  随机数
            int randomExpiryTime = 5 * 60 * 60 + RandomUtil.randomInt(60 * 60);
            redisTemplate.opsForValue().set(commentBestRedisKey, bestComment.getId(), randomExpiryTime, TimeUnit.SECONDS);

            // 返回最佳评论的 ID
            return bestComment.getId();
        }

        return null;
    }

    /**
     * 评论点赞
     *
     * @param likeCommentReqVO
     * @return
     */
    @Override
    public Response<?> likeComment(LikeCommentReqVO likeCommentReqVO) {

        // 被点赞的评论 ID
        Long commentId = likeCommentReqVO.getCommentId();

        // 1. 校验被点赞的评论是否存在
        checkCommentIsExist(commentId);

        // 2. 判断目标评论，是否已经被点赞
        // 当前登录用户
        Long userId = LoginUserContextHolder.getUserId();

        // rBitmap Key
        String rBitmapCommentLikesKey = RedisKeyConstants.buildRBitmapCommentLikesKey(userId);

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // Lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_comment_like_check.lua")));
        // 返回值类型
        script.setResultType(Long.class);

        // 执行 Lua 脚本，拿到返回结果
        Long result = redisTemplate.execute(script, Collections.singletonList(rBitmapCommentLikesKey), commentId);

        CommentLikeLuaResultEnum commentLikeLuaResultEnum = CommentLikeLuaResultEnum.valueOf(result);

        if (Objects.isNull(commentLikeLuaResultEnum)) {
            throw new BizException(ResponseCodeEnum.PARAM_NOT_VALID);
        }

        switch (commentLikeLuaResultEnum) {
            // Redis 中 rbitmap 不存在
            case NOT_EXIST -> {
                // 从数据库中校验评论是否被点赞，并异步初始化 rbitmap，设置过期时间
                int count = commentLikeDOMapper.selectCountByUserIdAndCommentId(userId, commentId);

                // 保底 1 小时 + 随机秒数
                long expireSeconds = 60 * 60 + RandomUtil.randomInt(60 * 60);

                // 目标评论已经被点赞
                if (count > 0) {
                    // 异步初始化 rbitmap
                    threadPoolTaskExecutor.submit(() ->
                            batchAddCommentLike2RBitmapAndExpire(userId, expireSeconds, rBitmapCommentLikesKey));

                    throw new BizException(ResponseCodeEnum.COMMENT_ALREADY_LIKED);
                }

                // 若目标评论没有被点赞，查询当前用户是否有点赞其他评论，有则同步初始化 rbitmap
                batchAddCommentLike2RBitmapAndExpire(userId, expireSeconds, rBitmapCommentLikesKey);

                // 添加当前点赞评论 ID 到 rbitmap 中
                // Lua 脚本路径
                script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_add_comment_like_and_expire.lua")));
                // 返回值类型
                script.setResultType(Long.class);
                redisTemplate.execute(script, Collections.singletonList(rBitmapCommentLikesKey), commentId, expireSeconds);
            }
            // 目标评论已经被点赞
            case COMMENT_LIKED -> throw new BizException(ResponseCodeEnum.COMMENT_ALREADY_LIKED);
        }

        // 3. 发送 MQ, 异步将评论点赞记录落库
        // 构建消息体 DTO
        LikeUnlikeCommentMqDTO likeUnlikeCommentMqDTO = LikeUnlikeCommentMqDTO.builder()
                .userId(userId)
                .commentId(commentId)
                // 点赞评论
                .type(LikeUnlikeCommentTypeEnum.LIKE.getCode())
                .createTime(LocalDateTime.now())
                .build();

        // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(likeUnlikeCommentMqDTO))
                .build();

        // 通过冒号连接, 可让 MQ 发送给主题 Topic 时，携带上标签 Tag
        String destination = MQConstants.TOPIC_COMMENT_LIKE_OR_UNLIKE + ":" + MQConstants.TAG_LIKE;

        // MQ 分区键
        String hashKey = String.valueOf(userId);

        // 异步发送 MQ 消息，提升接口响应速度
        rocketMQTemplate.asyncSendOrderly(destination, message, hashKey, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【评论点赞】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【评论点赞】MQ 发送异常: ", throwable);
            }
        });

        return Response.success();
    }

    /**
     * 取消评论点赞
     *
     * @param unLikeCommentReqVO
     * @return
     */
    @Override
    public Response<?> unlikeComment(UnLikeCommentReqVO unLikeCommentReqVO) {
        // 被取消点赞的评论 ID
        Long commentId = unLikeCommentReqVO.getCommentId();

        // 1. 校验评论是否存在
        checkCommentIsExist(commentId);

        Long userId = LoginUserContextHolder.getUserId();

        String rBitmapCommentLikesKey = RedisKeyConstants.buildRBitmapCommentLikesKey(userId);

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // Lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_comment_unlike_check.lua")));
        // 返回值类型
        script.setResultType(Long.class);
        // 执行 Lua 脚本，拿到返回结果
        Long result = redisTemplate.execute(script, Collections.singletonList(rBitmapCommentLikesKey), commentId);

        CommentUnlikeLuaResultEnum commentLikeLuaResultEnum = CommentUnlikeLuaResultEnum.valueOf(result);

        if (Objects.isNull(commentLikeLuaResultEnum)) {
            throw new BizException(ResponseCodeEnum.PARAM_NOT_VALID);
        }

        switch (commentLikeLuaResultEnum) {
            // Redis 中 rbitmap 不存在
            case NOT_EXIST -> {

                // 从数据库中校验评论是否被点赞
                int count = commentLikeDOMapper.selectCountByUserIdAndCommentId(userId, commentId);

                // 保底 1 小时 + 随机秒数
                long expireSeconds = 60 * 60 + RandomUtil.randomInt(60 * 60);

                // 未点赞，无法取消点赞操作，抛出业务异常
                if (count == 0) {
                    //  异步初始化 rbitmap
                    threadPoolTaskExecutor.submit(() -> batchAddCommentLike2RBitmapAndExpire(userId, expireSeconds, rBitmapCommentLikesKey));

                    throw new BizException(ResponseCodeEnum.COMMENT_NOT_LIKED);
                }

                // 若目标评论点赞了，查询当前用户是否有点赞其他评论，有则同步初始化 rbitmap
                batchAddCommentLike2RBitmapAndExpire(userId, expireSeconds, rBitmapCommentLikesKey);

                // Lua 脚本路径
                script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_add_comment_unlike_and_expire.lua")));
                // 返回值类型
                script.setResultType(Long.class);
                redisTemplate.execute(script, Collections.singletonList(rBitmapCommentLikesKey), commentId, expireSeconds);
            }
            // 评论没有被点赞
            case COMMENT_NOT_LIKED -> throw new BizException(ResponseCodeEnum.COMMENT_NOT_LIKED);
        }

        // 3. 发送顺序 MQ，删除评论点赞记录
        // 构建消息体 DTO
        LikeUnlikeCommentMqDTO likeUnlikeCommentMqDTO = LikeUnlikeCommentMqDTO.builder().userId(userId).commentId(commentId)
                // 取消点赞评论
                .type(LikeUnlikeCommentTypeEnum.UNLIKE.getCode()).createTime(LocalDateTime.now()).build();

        // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(likeUnlikeCommentMqDTO)).build();

        // 通过冒号连接, 可让 MQ 发送给主题 Topic 时，携带上标签 Tag
        String destination = MQConstants.TOPIC_COMMENT_LIKE_OR_UNLIKE + ":" + MQConstants.TAG_UNLIKE;

        // MQ 分区键
        String hashKey = String.valueOf(userId);

        // 异步发送 MQ 顺序消息，提升接口响应速度
        rocketMQTemplate.asyncSendOrderly(destination, message, hashKey, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【评论取消点赞】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【评论取消点赞】MQ 发送异常: ", throwable);
            }
        });

        return Response.success();
    }

    /**
     * 删除评论
     *
     * @param deleteCommentReqVO
     * @return
     */
    @Override
    public Response<?> deleteComment(DeleteCommentReqVO deleteCommentReqVO) {
        // 被删除的评论 ID
        Long commentId = deleteCommentReqVO.getCommentId();

        // 1. 校验评论是否存在
        CommentDO commentDO = commentDOMapper.selectByPrimaryKey(commentId);

        if (Objects.isNull(commentDO)) {
            throw new BizException(ResponseCodeEnum.COMMENT_NOT_FOUND);
        }

        // 2. 校验是否有权限删除
        Long currUserId = LoginUserContextHolder.getUserId();
        // RPC: 问题服务、查询问题作者信息
        Long quizCreatorId = quizRpcService.findQuizCreatorId(commentDO.getQuizId());

        if (!Objects.equals(currUserId, commentDO.getUserId()) && !Objects.equals(currUserId, quizCreatorId)) {
            throw new BizException(ResponseCodeEnum.COMMENT_CANT_OPERATE);
        }

        // 3. 物理删除评论、评论内容
        // 编程式事务，保证多个操作的原子性
        transactionTemplate.execute(status -> {
            try {
                // 删除评论元数据
                commentDOMapper.deleteByPrimaryKey(commentId);

                // 删除评论内容
                keyValueRpcService.deleteCommentContent(commentDO.getQuizId(),
                        commentDO.getCreateTime(),
                        commentDO.getContentUuid());

                return null;
            } catch (Exception ex) {
                // 标记事务为回滚
                status.setRollbackOnly();
                log.error("", ex);
                throw ex;
            }
        });

        // 4. 删除 Redis 缓存（ZSet 和 String）
        Long quizId = commentDO.getQuizId();

        String redisZSetKey = RedisKeyConstants.buildCommentListKey(quizId);

        // 使用 RedisTemplate 执行管道操作
        redisTemplate.executePipelined(new SessionCallback<>() {
            @Override
            public Object execute(RedisOperations operations) {

                // 如果该评论是最佳回答，则从 Redis 中删除该最佳评论缓存
                if (commentDO.getIsBest()) {
                    operations.delete(RedisKeyConstants.buildCommentBestKey(quizId));
                }

                // 删除 ZSet 中对应评论 ID
                operations.opsForZSet().remove(redisZSetKey, commentId);

                // 删除评论详情
                operations.delete(RedisKeyConstants.buildCommentDetailKey(commentId));
                return null;
            }
        });

        // 5. 发布广播 MQ, 将本地缓存删除
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_DELETE_COMMENT_LOCAL_CACHE, commentId, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【删除评论详情本地缓存】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【删除评论详情本地缓存】MQ 发送异常: ", throwable);
            }
        });

        // 6. 发送 MQ, 异步去更新计数、删除关联评论等
        // 构建消息对象，并将 DO 转成 Json 字符串设置到消息体中
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(commentDO))
                .build();

        // 异步发送 MQ 消息，提升接口响应速度
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_DELETE_COMMENT, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【评论删除】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【评论删除】MQ 发送异常: ", throwable);
            }
        });

        return Response.success();
    }

    /**
     * 删除本地评论缓存
     *
     * @param commentId
     */
    @Override
    public void deleteCommentLocalCache(Long commentId) {
        LOCAL_CACHE.invalidate(commentId);
    }

    /**
     * 设置评论为最佳
     *
     * @param bestCommentReqVO
     * @return
     */
    @Override
    public Response<?> bestComment(BestCommentReqVO bestCommentReqVO) {
        // 评论 ID
        Long commentId = bestCommentReqVO.getCommentId();
        // 1. 校验评论是否存在
        CommentDO commentDO = commentDOMapper.selectByPrimaryKey(commentId);
        if (Objects.isNull(commentDO)) {
            throw new BizException(ResponseCodeEnum.COMMENT_NOT_FOUND);
        }

        // 2. 校验该评论是否已经设置过最佳
        Preconditions.checkArgument(!commentDO.getIsBest(), ResponseCodeEnum.COMMENT_BEST_ALREADY_EXIST.getErrorMessage());

        // 3. 判断问题是否已经有最佳回答了
        // 先从 Redis 中查询是否已经有最佳回答的缓存
        Long quizId = commentDO.getQuizId();
        String commentBestRedisKey = RedisKeyConstants.buildCommentBestKey(quizId);

        // 保底 5 小时 + 随机秒数
        long expireSeconds = 5 * 60 * 60 + RandomUtil.randomInt(60 * 60);

        boolean hasKey = redisTemplate.hasKey(commentBestRedisKey);
        if (hasKey) {
            //  Redis 中有最佳回答的缓存，取出该值
            Object bestCommentId = redisTemplate.opsForValue().get(commentBestRedisKey);
            // 判断该值是否为空
            Preconditions.checkArgument(Objects.isNull(bestCommentId), ResponseCodeEnum.COMMENT_BEST_EXIST.getErrorMessage());
        } else {
            // Redis 中没有最佳回答的缓存，从数据库中查询该问题下是否已经有最佳回答
            CommentDO bestComment = commentDOMapper.selectBestCommentByQuizId(quizId);
            if (Objects.nonNull(bestComment)) {

                // 已经有最佳回答了，抛出异常，并将最佳回答的 ID 缓存到 Redis 中
                redisTemplate.opsForValue().set(commentBestRedisKey, bestComment.getId(), expireSeconds, TimeUnit.SECONDS);

                throw new BizException(ResponseCodeEnum.COMMENT_BEST_EXIST);
            }
        }

        // 4. 修改数据库的评论
        commentDOMapper.updateIsBest(commentId);

        // 5. 将评论 ID 从 Redis 中的 ZSet 列表剔除和删除对应的评论详情，并将该评论ID存入最佳评论中
        redisTemplate.executePipelined(new SessionCallback<>() {
            @Override
            public Object execute(RedisOperations operations) {
                // 删除 ZSet 中对应评论 ID
                String commentListRedisKey = RedisKeyConstants.buildCommentListKey(quizId);
                operations.opsForZSet().remove(commentListRedisKey, commentId);

                // 删除评论详情缓存
                String commentDetailRedisKey = RedisKeyConstants.buildCommentDetailKey(commentId);
                operations.delete(commentDetailRedisKey);

                // 添加到最佳评论
                String commentBestRedisKey = RedisKeyConstants.buildCommentBestKey(quizId);
                operations.opsForValue().set(commentBestRedisKey, commentId, expireSeconds, TimeUnit.SECONDS);
                return null;
            }
        });

        // 6. 发送 MQ，同步删除本地缓存中的评论详情
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_DELETE_COMMENT_LOCAL_CACHE, commentId, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【删除评论详情本地缓存】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【删除评论详情本地缓存】MQ 发送异常: ", throwable);
            }
        });
        return Response.success();
    }

    /**
     * 取消评论为最佳
     *
     * @param cancelBestCommentReqVO
     * @return
     */
    @Override
    public Response<?> cancelBestComment(CancelBestCommentReqVO cancelBestCommentReqVO) {
        // 评论 ID
        Long commentId = cancelBestCommentReqVO.getCommentId();
        // 1. 校验评论是否存在
        CommentDO commentDO = commentDOMapper.selectByPrimaryKey(commentId);
        if (Objects.isNull(commentDO)) {
            throw new BizException(ResponseCodeEnum.COMMENT_NOT_FOUND);
        }

        // 如果评论是最佳回答，则进行取消操作
        if (commentDO.getIsBest()) {
            // 问题 ID
            Long quizId = commentDO.getQuizId();
            String commentBestRedisKey = RedisKeyConstants.buildCommentBestKey(quizId);

            // 2. 修改数据库的评论
            commentDOMapper.updateIsBestCancel(commentId);

            // 3. 先使用 Lua 脚本将评论 ID 添加到 Redis 中的 ZSet 列表中，再删除对应的评论详情和删除该问题的最佳评论
            String commentLisRedisKey = RedisKeyConstants.buildCommentListKey(quizId);
            // 先使用 Lua 脚本将评论 ID 添加到 Redis 中的 ZSet 列表中
            DefaultRedisScript<Long> script = new DefaultRedisScript<>();
            // Lua 脚本路径
            script.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/cancel_best_comment.lua")));
            // Lua 返回值类型
            script.setResultType(Long.class);
            // 执行 Lua 脚本
            redisTemplate.execute(script, Collections.singletonList(commentLisRedisKey), commentId, commentId);

            // 删除对应的评论详情和删除该问题的最佳评论
            redisTemplate.executePipelined(new SessionCallback<>() {
                @Override
                public Object execute(RedisOperations operations) {
                    // 删除评论详情缓存
                    String commentDetailRedisKey = RedisKeyConstants.buildCommentDetailKey(commentId);
                    operations.delete(commentDetailRedisKey);

                    // 删除最佳评论
                    operations.delete(commentBestRedisKey);
                    return null;
                }
            });

            // 4. 发送 MQ，同步删除本地缓存中的评论详情
            rocketMQTemplate.asyncSend(MQConstants.TOPIC_DELETE_COMMENT_LOCAL_CACHE, commentId, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("==> 【删除评论详情本地缓存】MQ 发送成功，SendResult: {}", sendResult);
                }

                @Override
                public void onException(Throwable throwable) {
                    log.error("==> 【删除评论详情本地缓存】MQ 发送异常: ", throwable);
                }
            });
        } else {
            throw new BizException(ResponseCodeEnum.COMMENT_NOT_BEST);
        }

        return Response.success();
    }

    /**
     * 初始化评论点赞 RBitmap
     *
     * @param userId
     * @param expireSeconds
     * @param rBitmapCommentLikesKey
     * @return
     */
    private void batchAddCommentLike2RBitmapAndExpire(Long userId, long expireSeconds, String rBitmapCommentLikesKey) {
        try {
            // 查询该用户点赞的所有评论
            List<CommentLikeDO> commentLikeDOS = commentLikeDOMapper.selectByUserId(userId);

            // 若不为空，批量添加到 RBitmap 中
            if (CollUtil.isNotEmpty(commentLikeDOS)) {
                DefaultRedisScript<Long> script = new DefaultRedisScript<>();
                // lua 脚本路径
                script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_batch_add_comment_like_and_expire.lua")));
                // 返回值类型
                script.setResultType(Long.class);

                // 构建 Lua 参数
                List<Object> luaArgs = Lists.newArrayList();
                // 将每个点赞的评论 ID 传入
                commentLikeDOS.forEach(commentLikeDO -> luaArgs.add(commentLikeDO.getCommentId()));
                // 最后一个参数是过期时间（秒）
                luaArgs.add(expireSeconds);
                redisTemplate.execute(script, Collections.singletonList(rBitmapCommentLikesKey), luaArgs.toArray());
            }
        } catch (Exception e) {
            log.error("## 异步初始化【评论点赞】RBitmap异常：", e);
        }
    }

    /**
     * 校验被点赞的评论是否存在
     *
     * @param commentId
     */
    private void checkCommentIsExist(Long commentId) {
        // 先从本地缓存校验
        String localCacheJson = LOCAL_CACHE.getIfPresent(commentId);

        // 若本地缓存中，该评论不存在
        if (StringUtils.isBlank(localCacheJson)) {
            // 再从 Redis 中校验
            String commentDetailRedisKey = RedisKeyConstants.buildCommentDetailKey(commentId);

            boolean hasKey = redisTemplate.hasKey(commentDetailRedisKey);

            // 若 Redis 中也不存在
            if (!hasKey) {
                // 从数据库中校验
                CommentDO commentDO = commentDOMapper.selectByPrimaryKey(commentId);

                // 若数据库中，该评论也不存在，抛出业务异常
                if (Objects.isNull(commentDO)) {
                    throw new BizException(ResponseCodeEnum.COMMENT_NOT_FOUND);
                }
            }
        }
    }

    /**
     * 同步评论详情到本地缓存中
     *
     * @param findCommentItemRspVOS
     */
    private void syncCommentDetail2LocalCache(List<FindCommentItemRspVO> findCommentItemRspVOS) {
        // 开启一个异步线程
        threadPoolTaskExecutor.execute(() -> {
            // 构建缓存所需的键值
            Map<Long, String> localCacheData = Maps.newHashMap();
            findCommentItemRspVOS.forEach(findCommentItemRspVO -> {
                Long commentId = findCommentItemRspVO.getCommentId();
                localCacheData.put(commentId, JsonUtils.toJsonString(findCommentItemRspVO));
            });

            // 批量写入本地缓存
            LOCAL_CACHE.putAll(localCacheData);
        });
    }

    /**
     * 获取全部评论数据，并将评论详情同步到 Redis 中
     *
     * @param commentDOS
     * @param quizId
     * @return
     */
    private List<FindCommentItemRspVO> getCommentDataAndSync2Redis(List<CommentDO> commentDOS, Long quizId) {
        // 调用 KV 服务需要的入参
        List<FindCommentContentReqDTO> findCommentContentReqDTOS = Lists.newArrayList();
        // 调用用户服务的入参
        List<Long> userIds = Lists.newArrayList();

        // 循环提取 RPC 调用需要的入参数据
        commentDOS.forEach(commentDO -> {
            // 构建调用 KV 服务批量查询评论内容的入参
            Boolean isContentEmpty = commentDO.getIsContentEmpty();
            if (!isContentEmpty) {
                FindCommentContentReqDTO findCommentContentReqDTO = FindCommentContentReqDTO.builder().contentId(commentDO.getContentUuid()).yearMonth(DateConstants.DATE_FORMAT_Y_M.format(commentDO.getCreateTime())).build();
                findCommentContentReqDTOS.add(findCommentContentReqDTO);
            }

            // 构建调用用户服务的入参
            userIds.add(commentDO.getUserId());
            // 添加回复的评论用户 ID
            if (commentDO.getReplyUserId() != 0) {
                userIds.add(commentDO.getReplyUserId());
            }
        });

        // 并发查询优化
        // RPC: 调用 KV 服务
        CompletableFuture<List<FindCommentContentRspDTO>> KeyValueResultFuture = CompletableFuture.supplyAsync(() -> keyValueRpcService.batchFindCommentContent(quizId, findCommentContentReqDTOS), threadPoolTaskExecutor);

        // RPC: 调用用户服务
        CompletableFuture<List<FindUserByIdRspDTO>> userResultFuture = CompletableFuture.supplyAsync(() -> userRpcService.findByIds(userIds), threadPoolTaskExecutor);

        CompletableFuture<List<FindCommentItemRspVO>> resultFuture = CompletableFuture.allOf(KeyValueResultFuture, userResultFuture).thenApply(s -> {
            // 获取 Future 返回的结果
            List<FindCommentContentRspDTO> findCommentContentRspDTOS = KeyValueResultFuture.join();
            List<FindUserByIdRspDTO> findUserByIdRspDTOS = userResultFuture.join();

            // DTO 集合转 Map, 方便后续拼装数据
            Map<String, String> commentUuidAndContentMap = null;
            if (CollUtil.isNotEmpty(findCommentContentRspDTOS)) {
                commentUuidAndContentMap = findCommentContentRspDTOS.stream().collect(Collectors.toMap(FindCommentContentRspDTO::getContentId, FindCommentContentRspDTO::getContent));
            }

            Map<Long, FindUserByIdRspDTO> userIdAndUserMap = null;
            if (CollUtil.isNotEmpty(findUserByIdRspDTOS)) {
                userIdAndUserMap = findUserByIdRspDTOS.stream().collect(Collectors.toMap(FindUserByIdRspDTO::getId, dto -> dto));
            }

            List<FindCommentItemRspVO> findCommentItemRspVOS = Lists.newArrayList();

            // DO 转 VO
            for (CommentDO commentDO : commentDOS) {
                // 用户 ID
                Long userId = commentDO.getUserId();
                FindCommentItemRspVO findCommentItemRspVO = FindCommentItemRspVO.builder().userId(userId).commentId(commentDO.getId()).imageUrl(commentDO.getImageUrl()).createTime(DateUtils.formatRelativeTime(commentDO.getCreateTime())).isBest(commentDO.getIsBest()).build();

                // 用户信息
                setUserInfo(userIdAndUserMap, userId, findCommentItemRspVO);
                // 评论内容
                setCommentContent(commentUuidAndContentMap, commentDO, findCommentItemRspVO);
                // 回复用户信息
                setReplyUserInfo(userIdAndUserMap, commentDO.getReplyUserId(), findCommentItemRspVO);

                findCommentItemRspVOS.add(findCommentItemRspVO);
            }

            return findCommentItemRspVOS;
        });

        List<FindCommentItemRspVO> findCommentItemRspVOS = resultFuture.join();

        // 异步将评论详情，同步到 Redis 中
        threadPoolTaskExecutor.execute(() -> {
            // 准备批量写入的数据
            Map<String, String> data = Maps.newHashMap();
            findCommentItemRspVOS.forEach(findCommentItemRspVO -> {
                // 评论 ID
                Long commentId = findCommentItemRspVO.getCommentId();
                // 构建 Key
                String key = RedisKeyConstants.buildCommentDetailKey(commentId);
                data.put(key, JSON.toJSONString(findCommentItemRspVO));
            });

            // 使用 Redis Pipeline 提升写入性能
            redisTemplate.executePipelined(new SessionCallback<>() {
                @Override
                public Object execute(RedisOperations operations) {
                    for (Map.Entry<String, String> entry : data.entrySet()) {
                        // 随机生成过期时间 (5小时以内)
                        int randomExpire = RandomUtil.randomInt(5 * 60 * 60);
                        // 批量写入并设置过期时间
                        operations.opsForValue().set(entry.getKey(), entry.getValue(), randomExpire, TimeUnit.SECONDS);
                    }
                    return null;
                }
            });
        });

        return findCommentItemRspVOS;
    }

    /**
     * 设置回复用户信息
     *
     * @param userIdAndUserMap
     * @param replyUserId
     * @param findCommentItemRspVO
     */
    private static void setReplyUserInfo(Map<Long, FindUserByIdRspDTO> userIdAndUserMap, Long replyUserId, FindCommentItemRspVO findCommentItemRspVO) {
        // 如果存在回复的评论，则获取回复的评论用户信息
        if (replyUserId != 0) {
            // 获取回复的评论用户信息
            FindUserByIdRspDTO findUserByIdRspDTO = userIdAndUserMap.get(replyUserId);
            // 设置回复用户信息
            findCommentItemRspVO.setReplyUserId(replyUserId);
            findCommentItemRspVO.setReplyNickname(findUserByIdRspDTO.getDisplayName());
        }
    }

    /**
     * 同步评论至 Redis
     *
     * @param key
     * @param quizId
     */
    private void syncComments2Redis(String key, Long quizId) {
        // 获取最佳评论 ID
        Long bestCommentId = getBestCommentIdRedis(quizId);

        List<CommentDO> commentDOS = commentDOMapper.selectTop50Comments(quizId, bestCommentId);
        if (CollUtil.isNotEmpty(commentDOS)) {
            // 使用 Redis Pipeline 提升写入性
            redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                ZSetOperations<String, Object> zSetOps = redisTemplate.opsForZSet();

                // 遍历评论数并批量写入 ZSet
                for (CommentDO commentDO : commentDOS) {
                    Long commentId = commentDO.getId();
                    zSetOps.add(key, commentId, commentId);
                }

                // 设置随机过期时间，单位：秒
                // 5小时以内
                int randomExpiryTime = RandomUtil.randomInt(5 * 60 * 60);
                redisTemplate.expire(key, randomExpiryTime, TimeUnit.SECONDS);
                // 无返回值
                return null;
            });
        }
    }

    /**
     * 获取最佳评论并同步到 Redis 缓存
     *
     * @param quizId
     * @return
     */
    private Long getBestCommentIdRedis(Long quizId) {
        // 先判断该问题是否存在最佳评论
        // 先查询 redis
        String commentBestRedisKey = RedisKeyConstants.buildCommentBestKey(quizId);

        boolean hasKey = redisTemplate.hasKey(commentBestRedisKey);
        // 若存在，则从 Redis 中获取出最佳评论 ID
        Long bestCommentId = null;
        if (hasKey) {
            // 获取 Redis 中存储的 value
            Object redisValue = redisTemplate.opsForValue().get(commentBestRedisKey);

            if (Objects.nonNull(redisValue)) {
                bestCommentId = Long.parseLong(redisValue.toString());
            }
        }
        // Redis 不存在该数据，则从数据库中获取
        CommentDO bestCommentDO = commentDOMapper.selectBestCommentByQuizId(quizId);
        // 若数据库中存在该数据，则设置 bestCommentId 并，同步到redis中
        if (Objects.nonNull(bestCommentDO)) {
            bestCommentId = bestCommentDO.getId();
            // 随机设置过期时间 5 小时 + 随机秒数
            int randomExpiryTime = 5 * 60 * 60 + RandomUtil.randomInt(60 * 60);
            // 设置缓存
            redisTemplate.opsForValue().set(commentBestRedisKey, bestCommentId, randomExpiryTime, TimeUnit.SECONDS);
        }
        return bestCommentId;
    }

    /**
     * 设置评论内容
     *
     * @param commentUuidAndContentMap
     * @param commentDO
     * @param firstReplyCommentRspVO
     */
    private static void setCommentContent(Map<String, String> commentUuidAndContentMap, CommentDO commentDO, FindCommentItemRspVO firstReplyCommentRspVO) {
        if (CollUtil.isNotEmpty(commentUuidAndContentMap)) {
            String contentUuid = commentDO.getContentUuid();
            if (StringUtils.isNotBlank(contentUuid)) {
                firstReplyCommentRspVO.setContent(commentUuidAndContentMap.get(contentUuid));
            }
        }
    }

    /**
     * 设置用户信息
     *
     * @param userIdAndDTOMap
     * @param userId
     * @param findCommentItemRspVO
     */
    private static void setUserInfo(Map<Long, FindUserByIdRspDTO> userIdAndDTOMap, Long userId, FindCommentItemRspVO findCommentItemRspVO) {
        FindUserByIdRspDTO findUserByIdRspDTO = userIdAndDTOMap.get(userId);
        if (Objects.nonNull(findUserByIdRspDTO)) {
            findCommentItemRspVO.setAvatar(findUserByIdRspDTO.getPhotoUrl());
            findCommentItemRspVO.setNickname(findUserByIdRspDTO.getDisplayName());
        }
    }

    /**
     * 设置评论点赞数和是否点赞等信息
     *
     * @param findCommentItemRspVOS
     */
    private void setCommentCountAndLikeUnLikeInfo(List<FindCommentItemRspVO> findCommentItemRspVOS) {
        // 构建评论 ID 集合作为传参
        List<Long> commentIds = findCommentItemRspVOS.stream().map(FindCommentItemRspVO::getCommentId).toList();

        // 并行调用
        // 批量查询用户是否点赞
        CompletableFuture<Map<Long, Integer>> likedCommentRedisCacheFuture = CompletableFuture.supplyAsync(() ->
                rBitmapBatchCommentLikeCheck(commentIds), threadPoolTaskExecutor);

        // RPC：调用计数服务，查询评论点赞数
        CompletableFuture<List<FindCommentCountByIdRspDTO>> countFuture = CompletableFuture.supplyAsync(() ->
                countRpcService.findCommentsCountData(commentIds), threadPoolTaskExecutor);

        CompletableFuture<Object> result = CompletableFuture
                .allOf(likedCommentRedisCacheFuture, countFuture)
                .thenApply(s -> {
                    // 获取结果
                    Map<Long, Integer> commentIdAndIsLikedMap = likedCommentRedisCacheFuture.join();
                    List<FindCommentCountByIdRspDTO> findCommentCountByIdRspDTOS = countFuture.join();

                    if (CollUtil.isNotEmpty(findCommentCountByIdRspDTOS) && CollUtil.isNotEmpty(commentIdAndIsLikedMap)) {
                        // 转换为 Map
                        Map<Long, FindCommentCountByIdRspDTO> commentIdAndLikeTotalMap = findCommentCountByIdRspDTOS.stream()
                                .collect(Collectors.toMap(FindCommentCountByIdRspDTO::getCommentId, Function.identity()));

                        // 填充点赞数和用户是否点赞信息
                        findCommentItemRspVOS.forEach(findCommentItemRspVO -> {
                            // 评论 ID
                            Long commentId = findCommentItemRspVO.getCommentId();

                            Long likeTotal = commentIdAndLikeTotalMap.get(commentId).getLikeTotal();
                            Integer isLike = commentIdAndIsLikedMap.get(commentId);

                            findCommentItemRspVO.setLikeTotal(Objects.nonNull(likeTotal) ? likeTotal : 0L);
                            findCommentItemRspVO.setIsLiked(Objects.equals(LikeUnlikeCommentTypeEnum.LIKE.getCode(), isLike));
                        });
                    }
                    return null;
                });

        result.join();
    }

    /**
     * 批量查询用户是否点赞过该批评论
     *
     * @param commentIds
     * @return
     */
    private Map<Long, Integer> rBitmapBatchCommentLikeCheck(List<Long> commentIds) {
        // 查询登录用户是否点赞过该批评论
        // 获取当前登录用户 ID
        Long userId = LoginUserContextHolder.getUserId();
        // 构建 Redis Key
        String rBitmapCommentLikesRedisKey = RedisKeyConstants.buildRBitmapCommentLikesKey(userId);

        // 判断 RBitmap 是否存在
        boolean hasKey = redisTemplate.hasKey(rBitmapCommentLikesRedisKey);

        // 若不存在，则进行初始化
        if (!hasKey) {
            // 保底 1 小时 + 随机秒数
            long expireSeconds = 60 * 60 + RandomUtil.randomInt(60 * 60);

            batchAddCommentLike2RBitmapAndExpire(userId, expireSeconds, rBitmapCommentLikesRedisKey);
        }

        // 使用 Lua 脚本批量查询用户是否点赞过该批评论
        DefaultRedisScript<List> script = new DefaultRedisScript<>();
        // Lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_batch_comment_like_check.lua")));
        // 返回值类型
        script.setResultType(List.class);
        // 执行 Lua 脚本，拿到返回结果
        List<Object> result = redisTemplate.execute(script, Collections.singletonList(rBitmapCommentLikesRedisKey), commentIds.toArray());

        // 构建评论 ID 和是否点赞的映射
        Map<Long, Integer> commentIdAndIsLikedMap = Maps.newHashMap();

        if (CollUtil.isNotEmpty(result)) {
            for (int i = 0; i < commentIds.size(); i++) {
                // 获取评论 ID
                Long key = commentIds.get(i);
                // 获取点赞结果
                Object value = result.get(i);

                if (Objects.nonNull(value)) {
                    // 拼装到 commentIdAndIsLikedMap 中
                    commentIdAndIsLikedMap.put(
                            key,
                            Integer.parseInt(value.toString())
                    );
                }
            }
        }

        return commentIdAndIsLikedMap;
    }
}
